<?php
/*
 * client.php
 *
 * Copyright (C) 2012 - Dr.NP
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

/**
 * BSP client, PHP port
 * This client only support answering mode, we don't add any event listener and dispatcher at this time.
 * 
 * @package bsp
 * @author Dr.NP <np@bsgroup.org>
 * @update 05/21/2013
 * @changelog 
 * 		[08/15/2012] - Creation
 * 		[05/21/2013] - Rebuild with new protocol
 */

define('DEFAULT_BSP_SERVER_HOST', 		'localhost');
define('DEFAULT_BSP_SERVER_PORT', 		8254);

define('SOCK_TYPE_ANY', 				0);
define('SOCK_TYPE_TCP', 				1);
define('SOCK_TYPE_UDP', 				2);
define('INET_TYPE_ANY', 				0);
define('INET_TYPE_IPV4', 				1);
define('INET_TYPE_IPV6', 				2);
define('INET_TYPE_LOCAL', 				3);
define('DATA_TYPE_STREAM', 				0);
define('DATA_TYPE_PACKET', 				1);
define('PACKET_TYPE_REP', 				0);
define('PACKET_TYPE_RAW', 				1);
define('PACKET_TYPE_OBJ', 				2);
define('PACKET_TYPE_CMD', 				3);
define('PACKET_TYPE_HEARTBEAT', 		7);
define('COMPRESS_TYPE_NONE', 			0);
define('COMPRESS_TYPE_DEFLATE', 		1);
define('COMPRESS_TYPE_LZO', 			2);
define('COMPRESS_TYPE_SNAPPY', 			3);

define('READ_BLOCK_SIZE', 				4096);

$phpEx = substr(__FILE__, strrpos(__FILE__, '.') + 1);
require_once('packet.' . $phpEx);

class BspClient
{
	private $host;
	private $port;
	private $data_type;
	private $compress_type;
	private $mark_type;
	private $recv_buffer;
	private $send_buffer;
	
	private $sockfp = null;
	
	function __construct($inet = INET_TYPE_IPV4,
						$sock = SOCK_TYPE_TCP,
						$host = DEFAULT_BSP_SERVER_HOST,
						$port = DEFAULT_BSP_SERVER_PORT,
						$data_type = DATA_TYPE_STREAM)
	{
		$this->host = $host ? trim($host) : DEFAULT_BSP_SERVER_HOST;
		$this->port = $port ? $port : DEFAULT_BSP_SERVER_PORT;
		
		if (!function_exists('socket_create'))
		{
			die('<Socket> does not support' . PHP_EOL);
		}

		$domain = (INET_TYPE_IPV6 == $inet) ? AF_INET6 : AF_INET;
		$type = (SOCK_TYPE_UDP == $sock) ? SOCK_DGRAM : SOCK_STREAM;
		$protocol = (SOCK_TYPE_UDP == $sock) ? SOL_UDP : SOL_TCP;
		$this->sockfp = socket_create($domain, $type, $protocol);
		$this->data_type = (DATA_TYPE_STREAM == $data_type) ? DATA_TYPE_STREAM : DATA_TYPE_PACKET;
		$this->compress_type = COMPRESS_TYPE_NONE;
		$this->mark_type = PACKET_MARK_32B;
		
		return;
	}
	
	function __destruct()
	{
		if ($this->sockfp)
		{
			socket_close($this->sockfp);
		}
		
		return;
	}

	public function set_compress($comp = COMPRESS_TYPE_NONE)
	{
		$this->compress_type = (COMPRESS_TYPE_DEFLATE == $comp && function_exists('gzdeflate')) ? COMPRESS_TYPE_DEFLATE : COMPRESS_TYPE_NONE;

		return;
	}

	public function set_length_mark($mark = PACKET_MARK_32B)
	{
		$this->mark_type = (PACKET_MARK_64B == $mark) ? PACKET_MARK_64B : PACKET_MARK_32B;

		return;
	}
	
	// Blocking read data from socket
	private function _read()
	{
		if (!$this->sockfp)
		{
			return;
		}

		socket_set_nonblock($this->sockfp);
		while (1)
		{
			$tmp = socket_read($this->sockfp, READ_BLOCK_SIZE);
			if (false === $tmp)
			{
				// Error
				break;
			}

			if ('' === $tmp)
			{
				// Ending
				break;
			}
			$this->recv_buffer .= $tmp;
		}
		socket_set_block($this->sockfp);
		
		return;
	}

	private function _write($data)
	{
		if (!$this->sockfp)
		{
			return;
		}

		$sent = 0;
		$st = strlen($data);
		while (1)
		{
			$len = socket_write($this->sockfp, $data, $st);
			if (false == $len)
			{
				// Error
				break;
			}
			
			if ($len < $st)
			{
				$data = substr($data, $len);
				$st -= $len;
			}

			else
			{
				break;
			}
		}

		return;
	}

	private function _hdr($type)
	{
		$hdr = (($type & 0b111) << 5) | (($this->compress_type & 0b111) << 2) | ($this->mark_type & 0b11);
		
		return $hdr;
	}
	
	// TCP connector
	public function connect()
	{
		$this->recv_buffer = '';
		$this->send_buffer = '';
		return ($this->sockfp) ? socket_connect($this->sockfp, $this->host, $this->port) : false;
	}
	
	// Disconnect from server
	public function disconnect()
	{
		if ($this->sockfp)
		{
			socket_close($this->sockfp);
		}
		
		return;
	}
	
	// Binary stream
	public function send_stream($data = '')
	{
		if (DATA_TYPE_STREAM == $this->data_type)
		{
			$this->_write($data);
		}
		
		return;
	}
	
	// Raw data, input as a string
	public function send_raw($data = '')
	{
		if (DATA_TYPE_PACKET == $this->data_type)
		{
			$hdr = $this->_hdr(PACKET_TYPE_RAW);
			if (COMPRESS_TYPE_DEFLATE == $this->compress_type)
			{
				$data = gzdeflate($data);
			}
			$len = strlen($data);
			if (PACKET_MARK_32B == $this->mark_type)
			{
				$len_str = pack('N', $len);
			}

			else
			{
				$len_str = pack('NN', $len >> 32, $len & 0xFFFFFFFF);
			}
			
			$buf = chr($hdr) . $len_str . $data;
			$this->_write($buf);
		}
		
		return;
	}
	
	// Send object, which $obj is an object (Array)
	public function send_obj($obj = array())
	{
		if (DATA_TYPE_PACKET == $this->data_type)
		{
			$hdr = $this->_hdr(PACKET_TYPE_OBJ);
			if (PACKET_MARK_32B == $this->mark_type)
			{
				$pkg = new BspPacket();
				$data = $pkg->pack($obj);
				if (COMPRESS_TYPE_DEFLATE == $this->compress_type)
				{
					$data = gzdeflate($data);
				}
				$len = strlen($data);
				$len_str = pack('N', $len);
			}

			else
			{
				$pkg = new BspPacket(PACKET_MARK_64B);
				$data = $pkg->pack($obj);
				if (COMPRESS_TYPE_DEFLATE == $this->compress_type)
				{
					$data = gzdeflate($data);
				}
				$len = strlen($data);
				$len_str = pack('NN', $len >> 32, $len & 0xFFFFFFFF);
			}

			$buf = chr($hdr) . $len_str . $data;
			$this->_write($buf);
		}
		
		return;
	}
	
	// Send command, which $params is an object, $cmd is the command-id
	public function send_cmd($cmd, $params = array())
	{
		if (DATA_TYPE_PACKET == $this->data_type)
		{
			$hdr = $this->_hdr(PACKET_TYPE_CMD);
			if (PACKET_MARK_32B == $this->mark_type)
			{
				$pkg = new BspPacket();
				$data = pack('N', $cmd) . $pkg->pack($params);
				if (COMPRESS_TYPE_DEFLATE == $this->compress_type)
				{
					$data = gzdeflate($data);
				}
				$len = strlen($data);
				$len_str = pack('N', $len);
			}

			else
			{
				$pkg = new BspPacket(PACKET_MARK_64B);
				$data = pack('N', cmd) . $pkg->pack($obj);
				if (COMPRESS_TYPE_DEFLATE == $this->compress_type)
				{
					$data = gzdeflate($data);
				}
				$len = strlen($data);
				$len_str = pack('NN', $len >> 32, $len & 0xFFFFFFFF);
			}

			$buf = chr($hdr) . $len_str . $data;
			$this->_write($buf);
		}
		
		return;
	}
	
	public function recv_data()
	{
		$this->recv_buffer = '';
		if (DATA_TYPE_STREAM == $this->data_type)
		{
			// Return immediately
			$this->_read();
			return $this->recv_buffer;
		}

		else
		{
			// Read packet
			while (1)
			{
				$this->_read();
				if (strlen($this->recv_buffer) < 1)
				{
					continue;
				}

				// Parse header
				$hdr = ord($this->recv_buffer[0]);
				$pt = ($hdr >> 5) & 0b111;
				$pc = ($hdr >> 2) & 0b111;
				$pm = $hdr & 0b11;

				if (PACKET_MARK_32B == $pm)
				{
					if (strlen($this->recv_buffer) < 5)
					{
						continue;
					}

					$len = (ord($this->recv_buffer[1]) << 24) +
							(ord($this->recv_buffer[2]) << 16) +
							(ord($this->recv_buffer[3]) << 8) +
							ord($this->recv_buffer[4]);
					if (strlen($this->recv_buffer) < 5 + $len)
					{
						continue;
					}

					$data = substr($this->recv_buffer, 5, $len);
				}

				else
				{
					if (strlen($this->recv_buffer) < 9)
					{
						continue;
					}

					$len = (ord($this->recv_buffer[5]) << 24) +
							(ord($this->recv_buffer[6]) << 16) +
							(ord($this->recv_buffer[7]) << 8) +
							ord($this->recv_buffer[8]);
					if (strlen($this->recv_buffer) < 9 + $len)
					{
						continue;
					}

					$data = substr($this->recv_buffer, 9, $len);
				}

				// Decompress
				if (COMPRESS_TYPE_DEFLATE == $pc)
				{
					$data = gzinflate($data);
				}

				switch ($pt)
				{
					case PACKET_TYPE_RAW :
						$rtn = array('type' => $pt, 'raw' => $data);
						break;
					case PACKET_TYPE_OBJ :
						$pkg = new BspPacket($pm);
						$obj = $pkg->unpack($data);
						$rtn = array('type' => $pt, 'obj' => $obj);
						break;
					case PACKET_TYPE_CMD :
						$pkg = new BspPacket($pm);
						$cmd = (ord($data[0]) << 24) +
								(ord($data[1]) << 16) +
								(ord($data[2]) << 8) +
								ord($data[3]);
						$params = $pkg->unpack(substr($data, 4));
						$rtn = array('type' => $pt, 'cmd' => $cmd, 'params' => $params);
						break;
					default :
						// Unknown type
						$rtn = array('type' => $pt);
						break;
				}

				return $rtn;
			}
		}

		return null;
	}
}
?>
